package com.itc.bi.service.impl;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.itc.bi.bo.DataSetColumnTypeBO;
import com.itc.bi.config.ConnectionClickHouseConfig;
import com.itc.bi.constant.RedisConstant;
import com.itc.bi.constant.SystemConstant;
import com.itc.bi.entity.*;
import com.itc.bi.mapper.*;
import com.itc.bi.utils.DruidUtils;
import com.itc.common.core.exception.ServiceException;
import com.itc.common.core.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.sql.*;
import java.sql.Date;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @BelongsProject: itc-airport-cloud
 * @BelongsPackage: com.itc.bi.service.impl
 * @Author: hecaiy
 * @CreateTime: 2024-06-29  14:44
 * @Description: TODO 操作ClickHouse数据库相关信息
 * @Version: 1.0
 */
@Service
@Slf4j
public class ClickHouseJDBCService {

    @Autowired
    private ConnectionClickHouseConfig connectionClickHouseConfig;
    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private TableColumnMapper tableColumnMapper;
    @Autowired
    private TableMapper tableMapper;
    @Autowired
    private DataSetColumnMapper dataSetColumnMapper;
    @Autowired
    private DirectoryMapper directoryMapper;
    @Autowired
    private DataProcessMapper dataProcessMapper;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;

//    /**
//     * 获取连接
//     * @return
//     * @throws SQLException
//     */
//    public Connection getConn() throws SQLException {
//        //TODO 此处暂时这样，后期需要替换成连接池
//        DruidDataSource dataSource = new DruidDataSource();
//        dataSource.setDriverClassName(connectionClickHouseConfig.getDriverClassName());
//        dataSource.setUrl(connectionClickHouseConfig.getUrl());
//        dataSource.setUsername(connectionClickHouseConfig.getUsername());
//        dataSource.setPassword(connectionClickHouseConfig.getPassword());
//        Connection connection = dataSource.getConnection();
//        return connection;
//    }

//    public Connection getConn() {
//        ClickHouseProperties properties = new ClickHouseProperties();
//        properties.setUser(connectionClickHouseConfig.getUsername());
//        properties.setPassword(connectionClickHouseConfig.getPassword());
////        properties.setDatabase(db);
//        properties.setSocketTimeout(60000);
//        ClickHouseDataSource clickHouseDataSource = new ClickHouseDataSource(connectionClickHouseConfig.getUrl(), properties);
//        ClickHouseConnection conn = null;
//        try {
//            conn = clickHouseDataSource.getConnection();
//            System.out.println(conn);
//            System.out.println("连接成功");
//            return conn;
//        } catch (SQLException e) {
//            e.printStackTrace();
//        }
//        return null;
//    }

    /**
     * 池化技术
     * @return
     */
    public Connection getConn() {
        return DruidUtils.getConnection();
    }

    /**
     * 根据表名称  插入数据
     * @param tableName
     * @param dataList
     */
    public void insertDataByTableNameHashMap(String tableName,List<Map<String,Object>> dataList){
        if(ObjectUtil.isNotEmpty(dataList)){
            StringBuilder insertSql = new StringBuilder();
            insertSql.append("INSERT INTO `" + tableName + "` ( "+ SystemConstant.DATABASEREDUNDANCYCOLUMN+",");
            StringBuilder value = new StringBuilder();
            Boolean isTrue = false;
            value.append(" ) VALUES (?,");
            String sqlStr = "";
            Connection conn = null;
            PreparedStatement ps = null;
            try {
                conn =getConn();
                Map<String, Object> objectMap = dataList.get(0);
                //第一次循环拼接插入数据表头
                if (!isTrue) {
                    objectMap.forEach((key, value1) -> {
                        insertSql.append("`" + key + "`,");
                        value.append("?,");
                    });
                    value.deleteCharAt((value.lastIndexOf(",")));
                    insertSql.deleteCharAt((insertSql.lastIndexOf(",")));
                    isTrue = true;
                    sqlStr = insertSql.toString()+ value.toString()+")";
                }

                ps = conn.prepareStatement(sqlStr);
                for (int i = 0; i < dataList.size(); i++) {
                    Map<String, Object> stringObjectHashMap = dataList.get(i);
                    int k = 2;
                    ps.setObject(1, IdWorker.getId());
                    for (Map.Entry<String,Object> entry : stringObjectHashMap.entrySet()){
                        Object valueData = entry.getValue();
                        if(ObjectUtil.isEmpty(valueData)){
                            ps.setObject(k,null);
                        }else if(valueData instanceof DateTime){
                            //TODO hecaiyun测试出  hutool工具类导入数据的bug,这里有一个时间字段类型的bug,将时间字段类型为空的数据设置为null,单独处理，暂无更好的解决办法
                            ps.setObject(k, valueData.toString());
                        }else if(valueData instanceof Date){
                            ps.setObject(k,new Timestamp(((Date) valueData).getTime()));
                        }else if(valueData instanceof LocalDate){ //时间类型还可能存在其他类型
                            ps.setObject(k,Date.valueOf((LocalDate) entry.getValue()));
                        }else if(valueData instanceof LocalDateTime){
                            ps.setObject(k,Timestamp.valueOf((LocalDateTime) entry.getValue()));
                        }
                        else {
                            ps.setObject(k, valueData);
                        }
                        k++;
                    }
                    ps.addBatch();
                    //50条数据入库一次
                    if (i % 1000 == 0 ) {
                        ps.executeBatch();
                        ps.clearBatch();
                    }
                }
                //最后一次入库操作,避免数据丢失
                ps.executeBatch();
                ps.clearBatch();
                log.info("数据导入成功,tableName"+tableName+"********************************************");
            } catch (SQLException e) {
                log.error("数据导入失败,tableName"+tableName+"---------------------------");
                log.error(e.getMessage());
            }finally {
                closeConn(conn,ps,null,null);
            }
        }
    }

    /**
     * 根据表名称  插入数据
     * @param tableName
     * @param dataList
     */
    public void insertDataByTableHashMap(String tableName,List<Map<String,Object>> dataList){
        if(ObjectUtil.isNotEmpty(dataList)){
            StringBuilder insertSql = new StringBuilder();
            insertSql.append("INSERT INTO `" + tableName + "` ( ");
            StringBuilder value = new StringBuilder();
            Boolean isTrue = false;
            value.append(" ) VALUES (");
            String sqlStr = "";
            Connection conn = null;
            PreparedStatement ps = null;
            try {
                conn =getConn();
                Map<String, Object> objectMap = dataList.get(0);
                //第一次循环拼接插入数据表头
                if (!isTrue) {
                    objectMap.forEach((key, value1) -> {
                        insertSql.append("`" + key + "`,");
                        value.append("?,");
                    });
                    value.deleteCharAt((value.lastIndexOf(",")));
                    insertSql.deleteCharAt((insertSql.lastIndexOf(",")));
                    isTrue = true;
                    sqlStr = insertSql.toString()+ value.toString()+")";
                }

                ps = conn.prepareStatement(sqlStr);
                for (int i = 0; i < dataList.size(); i++) {
                    Map<String, Object> stringObjectHashMap = dataList.get(i);
                    int k = 1;
                    for (Map.Entry<String,Object> entry : stringObjectHashMap.entrySet()){
                        Object valueData = entry.getValue();
                        if(valueData instanceof DateTime){
                            ps.setObject(k, valueData.toString());
                        }else {
                            ps.setObject(k, valueData);
                        }
                        k++;
                    }
                    ps.addBatch();
                    //50条数据入库一次
                    if (i % 10000 == 0 ) {
                        ps.executeBatch();
                        ps.clearBatch();
                    }
                }
                //最后一次入库操作,避免数据丢失
                ps.executeBatch();
                ps.clearBatch();
            } catch (SQLException e) {
                throw new ServiceException("数据导入失败"+e.getMessage());
            }finally {
                closeConn(conn,ps,null,null);
            }
        }
    }

    /**
     * 根据表名称  插入数据
     * @param tableName
     * @param dataList
     */
    public void insertDataByTableNameHashMapOld(String tableName,List<Map<String,Object>> dataList){
        StringBuilder insertSql = new StringBuilder();
        insertSql.append("INSERT INTO `" + tableName + "` ( ");
        StringBuilder value = new StringBuilder();
        Boolean isTrue = false;
        int k = 0;

        for (int i = 0; i < dataList.size(); i++) {
            Map<String, Object> objectMap = dataList.get(i);
            for (Map.Entry<String,Object> entry : objectMap.entrySet()){
                if (!isTrue) {
                    insertSql.append("`" + entry.getKey() + "`,");
                }
                Object keyValue = entry.getValue();
                if(ObjectUtil.isNotEmpty(keyValue)){
                    value.append("'" + entry.getValue() + "',");
                }else {
                    value.append("'',");
                }
            }
            value.deleteCharAt((value.lastIndexOf(",")));
            if(!isTrue){
                insertSql.deleteCharAt((insertSql.lastIndexOf(",")));
                isTrue = true;
            }
            value.append("),(");
            k++;
            //50条数据入库一次
            if (k % 10000 == 0 ) {
                value.deleteCharAt((value.lastIndexOf("(")));
                value.deleteCharAt((value.lastIndexOf(",")));
                //数据入库操作
                String insertSqlStr = insertSql.toString() +" ) VALUES ( "+ value.toString();
                this.insertData(insertSqlStr);
                value.setLength(0);
                k = 0;
            }
        }
        //最后一次入库操作,避免数据丢失
        if (StringUtils.isNotEmpty(value.toString())) {
            value.deleteCharAt((value.lastIndexOf("(")));
            value.deleteCharAt((value.lastIndexOf(",")));
            //数据入库操作
            String insertSqlStr = insertSql.toString() + " ) VALUES ( "+ value.toString();
            this.insertData(insertSqlStr);
            value.setLength(0);
        }
    }

    /**
     * 根据表名称  插入数据
     * @param tableName
     * @param dataList
     */
//    @Async("asyncServiceExecutor")
    public void insertDataByTableNameObject(String tableName,List<Object> dataList){
        if(ObjectUtil.isNotEmpty(dataList)){
            StringBuilder insertSql = new StringBuilder();
            insertSql.append("INSERT INTO `" + tableName + "` ( "+ SystemConstant.DATABASEREDUNDANCYCOLUMN+",");
            StringBuilder value = new StringBuilder();
            Boolean isTrue = false;
            value.append(" ) VALUES (?,");
            String sqlStr = "";
            Connection conn = null;
            PreparedStatement ps = null;
            try {
                conn =getConn();
                Object object = dataList.get(0);
                if(object instanceof HashMap){
                    Map<String, Object> objectMap = (HashMap)object;
                    //第一次循环拼接数据
                    if (!isTrue) {
                        objectMap.forEach((key, value1) -> {
                            insertSql.append("`" + key + "`,");
                            value.append("?,");
                        });
                        value.deleteCharAt((value.lastIndexOf(",")));
                        insertSql.deleteCharAt((insertSql.lastIndexOf(",")));
                        isTrue = true;
                        sqlStr = insertSql.toString()+ value.toString()+")";
                    }

                    ps = conn.prepareStatement(sqlStr);
                    for (int i = 0; i < dataList.size(); i++) {
                        Map<String, Object> stringObjectHashMap = (HashMap)dataList.get(i);
                        int k = 2;
                        ps.setObject(1, IdWorker.getId());
                        for (Map.Entry<String,Object> entry : stringObjectHashMap.entrySet()){
                            Object objectValue = entry.getValue();
                            if(ObjectUtil.isEmpty(objectValue)){
                                ps.setObject(k,null);
                            }else if(objectValue instanceof Date){
                                ps.setObject(k,new Timestamp(((Date) objectValue).getTime()));
                            }
                            else if(objectValue instanceof LocalDate){
                                ps.setObject(k,Date.valueOf((LocalDate) entry.getValue()));
                            }else if(objectValue instanceof LocalDateTime){
                                ps.setObject(k,Timestamp.valueOf((LocalDateTime) entry.getValue()));
                            }else {
                                ps.setObject(k, entry.getValue());
                            }
                            k++;
                        }
                        ps.addBatch();
                        //50条数据入库一次
                        if (i % 10000 == 0 ) {
                            ps.executeBatch();
                            ps.clearBatch();
                        }
                    }
                }
                if (object instanceof JSONObject){
                    JSONObject jsonObject = JSONObject.parseObject(JSON.toJSONString(object));
                    //第一次循环拼接数据
                    if (!isTrue) {
                        jsonObject.entrySet().forEach(entry -> {
                            String key = entry.getKey();
                            insertSql.append("`" + key + "`,");
                            value.append("?,");
                        });
                        value.deleteCharAt((value.lastIndexOf(",")));
                        insertSql.deleteCharAt((insertSql.lastIndexOf(",")));
                        isTrue = true;
                        sqlStr = insertSql.toString()+ value.toString()+")";
                    }

                    ps = conn.prepareStatement(sqlStr);
                    for (int i = 0; i < dataList.size(); i++) {
                        JSONObject stringObjectHashMap = JSONObject.parseObject(JSON.toJSONString(object));
                        int k = 2;
                        ps.setObject(1, IdWorker.getId());
                        Set<Map.Entry<String, Object>> entries = jsonObject.entrySet();
                        for (Map.Entry<String, Object> entry : entries) {
                            Object objectValue = entry.getValue();
                            if(objectValue instanceof LocalDate){
                                ps.setObject(k,Date.valueOf((LocalDate) entry.getValue()));
                            }else if(objectValue instanceof Date){
                                ps.setObject(k,new Timestamp(((Date) objectValue).getTime()));
                            }else if(objectValue instanceof LocalDateTime){
                                ps.setObject(k,Timestamp.valueOf((LocalDateTime) entry.getValue()));
                            }else {
                                ps.setObject(k, entry.getValue());
                            }
                            k++;
                        }
                        ps.addBatch();
                        //50条数据入库一次
                        if (i % 10000 == 0 ) {
                            ps.executeBatch();
                            ps.clearBatch();
                        }
                    }
                }

                //最后一次入库操作,避免数据丢失
                ps.executeBatch();
                ps.clearBatch();
            } catch (SQLException e) {
                throw new ServiceException("数据导入失败"+e.getMessage());
            }finally {
                closeConn(conn,ps,null,null);
            }
        }
    }

    /**
     * 根据表名称  插入数据
     * @param tableName
     * @param dataList
     */
    @Async("asyncServiceExecutor")
    public void insertDataByTableNameObjectOld(String tableName,List<Object> dataList){
        StringBuilder insertSql = new StringBuilder();
        insertSql.append("INSERT INTO `" + tableName + "` ( ");
        StringBuilder value = new StringBuilder();
        Boolean isTrue = false;
        int k = 0;
        for (int i = 0; i < dataList.size(); i++) {
            Object object = dataList.get(i);
            Map<String, Object> objectMap = (HashMap)object;
            for (Map.Entry<String,Object> entry : objectMap.entrySet()){
                if (!isTrue) {
                    insertSql.append("`" + entry.getKey() + "`,");
                }
                Object keyValue = entry.getValue();
                if(ObjectUtil.isNotEmpty(keyValue)){
                    value.append("'" + entry.getValue() + "',");
                }else {
                    value.append("'',");
                }
            }
            value.deleteCharAt((value.lastIndexOf(",")));
            if(!isTrue){
                insertSql.deleteCharAt((insertSql.lastIndexOf(",")));
                isTrue = true;
            }
            value.append("),(");
            k++;
            //50条数据入库一次
            if (k % 10000 == 0 ) {
                value.deleteCharAt((value.lastIndexOf("(")));
                value.deleteCharAt((value.lastIndexOf(",")));
                //数据入库操作
                String insertSqlStr = insertSql.toString() +" ) VALUES ( "+ value.toString();
                this.insertData(insertSqlStr);
                value.setLength(0);
                k = 0;
            }
        }
        //最后一次入库操作,避免数据丢失
        if (StringUtils.isNotEmpty(value.toString())) {
            value.deleteCharAt((value.lastIndexOf("(")));
            value.deleteCharAt((value.lastIndexOf(",")));
            //数据入库操作
            String insertSqlStr = insertSql.toString() + " ) VALUES ( "+ value.toString();
            this.insertData(insertSqlStr);
            value.setLength(0);
        }
    }

    /**
     * TODO 不能使用
     * @param tableName       数仓表名称
     * @param tbId            数据集id
     */
    public void  createTableByDsId(String tableName,Long tbId){
        List<TableColumnEntity> list = tableColumnMapper.selectList(Wrappers.<TableColumnEntity>lambdaQuery()
                .eq(TableColumnEntity::getTableId, tbId)
                .orderByAsc(TableColumnEntity::getColumnPosition));
        if(ObjectUtil.isEmpty(list)){
            log.error("数据集字段不能为空");
            throw new ServiceException("数据集字段不能为空");
        }
        String sql = "create table IF NOT EXISTS "+ "tableName" +"("+ SystemConstant.DATABASEREDUNDANCYCOLUMN +" Int64, ";
        for (TableColumnEntity tableColumnEntity : list) {
            String columnType = tableColumnEntity.getColumnType();
            if(ObjectUtil.isNotEmpty(columnType)){
                sql += tableColumnEntity.getColumnName() +  getColumnNameType(columnType) + ",";
            }
        }
        sql = sql.substring(0,sql.length()-1);
        sql += ")engine=MergeTree() order by "+SystemConstant.DATABASEREDUNDANCYCOLUMN;
        this.createTable(sql);
    }

    /**
     *
     * @param tableName      目标表名称
     * @param dtId            数据集id
     * @param orderByColumn   排序字段
     */
    public Boolean createTableSourceDS(String tableName,Long dtId,String orderByColumn){
        List<DataSetColumnEntity> list = dataSetColumnMapper.selectList(Wrappers.<DataSetColumnEntity>lambdaQuery()
                .eq(DataSetColumnEntity::getDataSetId, dtId)
                .orderByAsc(DataSetColumnEntity::getColumnPosition));
        if(ObjectUtil.isEmpty(list)){
            return false;
        }
        //TODO 暂时先用第一个字段作为排序字段 后续调整为取table中的默认值是否weinull
        if(ObjectUtil.isNotEmpty(list)){
            try{
                String columnName = list.get(0).getColumnName();
                return createTableByColumnList(tableName,list,columnName);
            }catch (Exception e){
                //创表格失败 返回false
                return false;
            }
        }
        return false;
    }

    /**
     * 获取字段类型
     * @param columnType
     * @return
     */
    public String getColumnNameTypeNotNUll(String columnType){
        if(columnType.equalsIgnoreCase("int") || columnType.equalsIgnoreCase("long")){
            return " Int64 ";
        }
        if(columnType.equalsIgnoreCase("string") ){
            return " String ";
        }
        if(columnType.equalsIgnoreCase("double") || columnType.equalsIgnoreCase("decimal") || columnType.equalsIgnoreCase("number")){
            return " Float64 ";
        }
        if (columnType.equalsIgnoreCase("number")){
            return " Float64 ";
        }
//        if(columnType.equalsIgnoreCase("decimal")){
//            return " Nullable(Decimal(64,2)) DEFAULT NULL ";
//        }
        if(columnType.equalsIgnoreCase("date")){
            return " Date ";
        }
        if(columnType.equalsIgnoreCase("timestamp")){
            return " DateTime ";
        }
        return " String ";
    }

    /**
     * 获取字段类型转置
     * @param columnType
     * @return
     */
    public String getColumnNameTypeTrs(String columnType){
        if(columnType.equalsIgnoreCase("Int64") || columnType.toLowerCase().contains("int")){
            return "number";
        }
        if(columnType.equalsIgnoreCase("Float64") || columnType.toLowerCase().contains("float") ){
            return "number";
        }
        if (columnType.equalsIgnoreCase("Float32")){
            return "number";
        }
        if(columnType.equalsIgnoreCase("Date") || columnType.toLowerCase().contains("date")){
            return "date";
        }
        if(columnType.equalsIgnoreCase("DateTime")){
            return "date";
        }
        return "string";
    }

    /**
     * 获取字段类型
     * @param columnType
     * @return
     */
    public String getColumnNameType(String columnType){
        if(columnType.equalsIgnoreCase("int") ){
            return " Nullable(Float64) DEFAULT NULL ";
        }
        if(columnType.equalsIgnoreCase("string") ){
            return " Nullable(String) DEFAULT NULL ";
        }
        if(columnType.equalsIgnoreCase("double") || columnType.equalsIgnoreCase("decimal") || columnType.equalsIgnoreCase("long") || columnType.equalsIgnoreCase("number")){
            return " Nullable(Float64) DEFAULT NULL ";
        }
        if (columnType.equalsIgnoreCase("float")){
            return " Nullable(Float64) DEFAULT NULL ";
        }
//        if(columnType.equalsIgnoreCase("decimal")){
//            return " Nullable(Decimal(64,2)) DEFAULT NULL ";
//        }
        if(columnType.equalsIgnoreCase("date")){
            return " Nullable(DateTime) DEFAULT NULL ";
        }
        if(columnType.equalsIgnoreCase("timestamp")){
            return " Nullable(DateTime) DEFAULT NULL ";
        }
        return " Nullable(String) DEFAULT NULL ";
    }

    /**
     * 根据数据列创建表
     * @param tableName
     * @param list
     * @return
     */
    public Boolean createTableByColumnList(String tableName,List<DataSetColumnEntity> list,String orderByColumn){
        String sql = getCreateTableByColumnListSql(tableName, list, orderByColumn);
//        String sql = "create table IF NOT EXISTS "+ tableName +"("+ SystemConstant.DATABASEREDUNDANCYCOLUMN +" Int64, ";
//        String columnOrderBy = SystemConstant.DATABASEREDUNDANCYCOLUMN;
//        if(ObjectUtil.isNotEmpty(orderByColumn)){
//            DataSetColumnEntity setColumnEntity = list.stream().filter(p -> p.getColumnName().equalsIgnoreCase(orderByColumn)).findFirst().orElse(null);
//            sql = "create table IF NOT EXISTS "+ tableName +"( "+setColumnEntity.getColumnName()+ getColumnNameTypeNotNUll(setColumnEntity.getColumnType()) + ",";
//            list.removeIf(dataSetColumnEntity -> dataSetColumnEntity.getColumnName().equalsIgnoreCase(orderByColumn));
//            columnOrderBy = orderByColumn;
//        }
//
//        for (DataSetColumnEntity tableColumnEntity : list) {
//            String columnType = tableColumnEntity.getColumnType();
//            if(ObjectUtil.isNotEmpty(columnType)){
//                sql += tableColumnEntity.getColumnName() +  getColumnNameType(columnType) + ",";
//            }
//        }
//        sql = sql.substring(0,sql.length()-1);
//        sql += ")engine=MergeTree() order by "+columnOrderBy;
//TODO 已弃用  留存
//        String sql = "create table IF NOT EXISTS "+ tableName +"(";
//        for (int i = 0; i < list.size(); i++) {
//            DataSetColumnEntity tableColumnEntity = list.get(i);
//            String characterType = tableColumnEntity.getColumnType();
//            if (i==0){
//                if(characterType.contains("string") || characterType.contains("date")){
//                    characterType = "String";
//                }else {
//                    characterType = "Decimal(64,2)";
//                }
//            }else {
//                if(characterType.contains("string") || characterType.contains("date")){
//                    characterType = "Nullable(String)";
//                }else {
//                    characterType = "Nullable(Decimal(64,2))";
//                }
//            }
//            if(tableColumnEntity.getColumnName().equals(orderByColumn)){
//                if(characterType.contains("string") || characterType.contains("date")){
//                    characterType = "String";
//                }else {
//                    characterType = "Decimal(64,2)";
//                }
//            }
//            sql += tableColumnEntity.getColumnName() + " " + characterType + ",";
//        }
//        if(ObjectUtil.isEmpty(orderByColumn)){
//            orderByColumn = list.get(0).getColumnName();
//        }
//        sql = sql.substring(0,sql.length()-1);
//        sql += ")engine=MergeTree() order by "+orderByColumn;
        log.info("sql:"+sql);
        this.createTable(sql);
        return true;
    }

    /**
     * 根据数据列创建表
     * @param tableName
     * @param list
     * @param orderByColumn
     * @return
     */
    public String getCreateTableByColumnListSql(String tableName,List<DataSetColumnEntity> list,String orderByColumn){
        String sql = "create table IF NOT EXISTS "+ tableName +"("+ SystemConstant.DATABASEREDUNDANCYCOLUMN +" Int64, ";
        String columnOrderBy = SystemConstant.DATABASEREDUNDANCYCOLUMN;
        if(ObjectUtil.isNotEmpty(orderByColumn)){
            DataSetColumnEntity setColumnEntity = list.stream().filter(p -> p.getColumnName().equalsIgnoreCase(orderByColumn)).findFirst().orElse(null);
            sql = "create table IF NOT EXISTS "+ tableName +"( `"+setColumnEntity.getColumnName()+"`"+ getColumnNameTypeNotNUll(setColumnEntity.getColumnType()) + ",";
            list.removeIf(dataSetColumnEntity -> dataSetColumnEntity.getColumnName().equalsIgnoreCase(orderByColumn));
            columnOrderBy = orderByColumn;
        }

        for (DataSetColumnEntity tableColumnEntity : list) {
            String columnType = tableColumnEntity.getColumnType();
            if(ObjectUtil.isNotEmpty(columnType)){
                sql += "`"+tableColumnEntity.getColumnName()+"`" +  getColumnNameType(columnType) + ",";
            }
        }
        sql = sql.substring(0,sql.length()-1);
        sql += ")engine=MergeTree() order by `"+columnOrderBy+"`";
        return sql;
    }

    /**
     * 逐步加工生成数据表
     * @param tableName 生成的数据库表名称
     * @param sql       sql语句
     */
    public void insertTableProcessDS(String tableName,String sql){
        String createSql = " insert into "+tableName +" " + sql;
        this.execSql(createSql);
    }

    /**
     * 逐步加工生成数据表
     * @param tableName 生成的数据库表名称
     * @param dtId      原始数据集id  用于查询数据表字段
     * @param sql       sql语句
     */
    public void createTableProcessDS(String tableName,Long dtId,String sql){
        String orderByColumn = getDefaultColumn(sql);
        String createSql = "create table IF NOT EXISTS "+ tableName +
                " engine=MergeTree() " +
                " order by  `"+ orderByColumn +
                "` as " +
                "( "+sql+")";
        this.createTable(createSql);
    }

    /**
     * 获取默认的排序字段
     * @param dtId
     * @return
     */
    private String getDefaultColumn(String sql){
        String orderByColumn = "";
        String sqlStr = "select * from ( "+sql+" ) limit 1";
        List<LinkedHashMap<String,Object>> list = new ArrayList<>();
        Connection conn = null;
        Statement statement = null;
        try {
            conn = getConn();
            statement = conn.createStatement();
            ResultSet rs  = statement.executeQuery(sqlStr);
            ResultSetMetaData rsmd = rs.getMetaData();
            int columnCount = rsmd.getColumnCount();
            if(columnCount<=0){
                throw new ServiceException("sql 查询不到数据项 请检查");
            }
            //取第一个数据项返回
            return rsmd.getColumnName(1);
        }catch (SQLException e){
            log.error("查询数据失败"+ sql);
            throw new ServiceException("查询数据失败");
        }finally {
            closeConn(conn,  null, null,statement);
            //return orderByColumn;
        }
    }

    /**
     * 创建表
     * @param sql
     */
    public void createTable(String sql){
        RLock lock = redissonClient.getLock(RedisConstant.PERMISSION_LOCK);
        lock.lock();
        Connection conn = null;
        PreparedStatement pstmt = null;
        try {
            conn = getConn();
            pstmt = conn.prepareStatement(sql);
            pstmt.execute();
        } catch (SQLException e) {
            log.error("创建表失败"+e.getMessage());
            log.error("sql:"+sql);
            throw new ServiceException("创建表失败");
        }finally {
            lock.unlock();
            closeConn(conn, pstmt, null,null);
        }
    }

    /**
     * 执行sql  返回sql  是否能成功执行
     * @param sql
     */
    public Boolean execSql(String sql){
        Connection conn = null;
        PreparedStatement pstmt = null;
        try {
            conn = getConn();
            pstmt = conn.prepareStatement(sql);
            pstmt.execute();
            return true;
        } catch (SQLException e) {
            log.error("sql执行失败"+sql);
            return false;
        }finally {
            closeConn(conn, pstmt, null,null);
        }
    }

    /**
     * 获取数据列
     * @param tableName
     * @param colName
     * @return
     */
    public List<String> getColumnListByNameAndCol(String tableName,String colName){
        List<String> list = new ArrayList<>();
        String sql = "select `" + colName + "` from " + tableName +" group by `" + colName+"`";
        Connection conn = null;
        Statement statement = null;
        try {
            conn = getConn();
            statement = conn.createStatement();
            ResultSet rs  = statement.executeQuery(sql);
            ResultSetMetaData rsmd = rs.getMetaData();
            while (rs.next()){
                for (int i = 1; i <= rsmd.getColumnCount(); i++) {
                    Object object = rs.getObject(i);
                    if(ObjectUtil.isNotEmpty(object)){
                        list.add(object.toString());
                    }
                }
            }
        }catch (SQLException e){
            log.error("查询数据失败"+ sql);
            throw new ServiceException("查询数据失败");
        }finally {
            closeConn(conn,  null, null,statement);
        }
        return list;
    }

    /**
     * 不用分页 获取全部数据
     * @param tableName 表名称
     * @param param     过滤条件 where a=123465
     * @return
     */
    public List<LinkedHashMap<String,Object>> queryTableByTable(String tableName,String param){
        if(ObjectUtil.isEmpty(param)){
            param = "";
        }
        List<LinkedHashMap<String,Object>> list = new ArrayList<>();
        /**
         * select * from aaa where 1=1 limit 1,10
         */
        String sql = "select * from "+tableName + param;
        Connection conn = null;
        Statement statement = null;
        try {
            conn = getConn();
            statement = conn.createStatement();
            ResultSet rs  = statement.executeQuery(sql);
            ResultSetMetaData rsmd = rs.getMetaData();
            while (rs.next()){
                LinkedHashMap<String,Object> map = new LinkedHashMap<>();
                for (int i = 1; i <= rsmd.getColumnCount(); i++) {
                    Object valueObj = rs.getObject(i);
                    valueObj = formatDataValue(valueObj);
                    map.put(rsmd.getColumnName(i),valueObj);
                }
                list.add(map);
            }
        }catch (SQLException e){
            log.error("查询数据失败", e.getMessage());
            throw new ServiceException("查询数据失败");
        }finally {
            closeConn(conn,  null, null,statement);
        }
        //移除多余列
        removeColumn(list);
        return list;
    }

    /**
     * 获取表数据大小
     * @param sqlStr
     * @return
     */
    public Long getTableSizeSql(String sqlStr){
        String sql = "select count(*) as count from ("+sqlStr+")";
        List<LinkedHashMap<String,Object>> list = new ArrayList<>();
        Connection conn = null;
        Statement statement = null;
        try {
            conn = getConn();
            statement = conn.createStatement();
            ResultSet rs  = statement.executeQuery(sql);
            ResultSetMetaData rsmd = rs.getMetaData();
            while (rs.next()){
                LinkedHashMap<String,Object> map = new LinkedHashMap<>();
                for (int i = 1; i <= rsmd.getColumnCount(); i++) {
                    map.put(rsmd.getColumnName(i),rs.getObject(i));
                }
                list.add(map);
            }
            return Long.valueOf(list.get(0).get("count").toString());
        }catch (SQLException e){
            log.error("查询数据失败"+ sql);
        }finally {
            closeConn(conn,  null, null,statement);
        }
        return null;
    }

    /**
     * 获取表数据大小
     * @param tableName
     * @return
     */
    public Long getTableSizeTableName(String tableName){
        String sql = "select count(*) as count from "+tableName;
        List<LinkedHashMap<String,Object>> list = new ArrayList<>();
        Connection conn = null;
        Statement statement = null;
        try {
            conn = getConn();
            statement = conn.createStatement();
            ResultSet rs  = statement.executeQuery(sql);
            ResultSetMetaData rsmd = rs.getMetaData();
            while (rs.next()){
                LinkedHashMap<String,Object> map = new LinkedHashMap<>();
                for (int i = 1; i <= rsmd.getColumnCount(); i++) {
                    map.put(rsmd.getColumnName(i),rs.getObject(i));
                }
                list.add(map);
            }
            return Long.valueOf(list.get(0).get("count").toString());
        }catch (SQLException e){
            log.error("查询数据失败"+ sql);
        }finally {
            closeConn(conn,  null, null,statement);
        }
        return null;
    }

    /**
     * 根据表名查询数据
     * @param tableName
     * @param pageIndex
     * @param pageSize
     * @return
     */
    public List<LinkedHashMap<String,Object>> queryTableByTable(String tableName,Integer pageIndex,Integer pageSize,String param){
        if(pageIndex<=0){
            pageIndex = 1;
        }
        if(pageSize<=0){
            pageSize = 10;
        }
        /**
         * select * from aaa where 1=1 limit 1,10
         */
        String sql = "select * from "+tableName + param +" limit "+(pageIndex-1)*pageSize+","+pageSize;
        List<LinkedHashMap<String,Object>> list = new ArrayList<>();
        Connection conn = null;
        Statement statement = null;
        try {
            conn = getConn();
            statement = conn.createStatement();
            ResultSet rs  = statement.executeQuery(sql);
            ResultSetMetaData rsmd = rs.getMetaData();
            while (rs.next()){
                LinkedHashMap<String,Object> map = new LinkedHashMap<>();
                for (int i = 1; i <= rsmd.getColumnCount(); i++) {
                    if(!rsmd.getColumnName(i).equalsIgnoreCase(SystemConstant.DATABASEREDUNDANCYCOLUMN)){
                        Object valueObj = rs.getObject(i);
                        valueObj = formatDataValue(valueObj);
                        map.put(rsmd.getColumnName(i),valueObj);
                    }
                }
                list.add(map);
            }
        }catch (SQLException e){
            log.error("查询数据失败"+ sql);
            throw new ServiceException("查询数据失败");
        }finally {
            closeConn(conn,  null, null,statement);
        }
        //removeColumn(list);
        return list;
    }

    /**
     * 时间格式化
     * @param value
     * @return
     */
    public Object formatDataValue(Object value){
        if(value instanceof Date){
            return DateUtil.format((Date) value, DatePattern.NORM_DATE_PATTERN);
        }
        if(value instanceof Timestamp){
            return DateUtil.format((Timestamp) value, DatePattern.NORM_DATETIME_PATTERN);
        }
        return value;
    }

    /**
     * 根据查询sql返回字段类型
     * @param sql
     * @return
     */
    public List<DataSetColumnTypeBO> getColumnListBySql(String sql){
        List<DataSetColumnTypeBO> list = new ArrayList<>();
        Connection conn = null;
        Statement statement = null;
        try {
            conn = getConn();
            statement = conn.createStatement();
            ResultSet rs  = statement.executeQuery(sql);
            ResultSetMetaData rsmd = rs.getMetaData();
            for (int i = 1; i <= rsmd.getColumnCount(); i++) {
                list.add(new DataSetColumnTypeBO(rsmd.getColumnName(i),getColumnNameTypeTrs(rsmd.getColumnTypeName(i))));
            }
        }catch (SQLException e){
            log.error("查询数据失败"+ sql);
        }finally {
            closeConn(conn,  null, null,statement);
            return list;
        }
    }

    /**
     * 查询数据 验证sql语句
     * @param sql
     * @return
     */
    public JSONObject queryTableBySqlToSuccess(String sql){
        JSONObject jsonObject = new JSONObject();
        List<LinkedHashMap<String,Object>> list = new ArrayList<>();
        jsonObject.put("body",list);
        Connection conn = null;
        Statement statement = null;
        try {
            conn = getConn();
            statement = conn.createStatement();
            ResultSet rs  = statement.executeQuery(sql);
            ResultSetMetaData rsmd = rs.getMetaData();
            while (rs.next()){
                LinkedHashMap<String,Object> map = new LinkedHashMap<>();
                for (int i = 1; i <= rsmd.getColumnCount(); i++) {
                    map.put(rsmd.getColumnName(i),rs.getObject(i));
                }
                list.add(map);
            }
            removeColumn(list);
            jsonObject.put("body",list);
            jsonObject.put("success",true);
            jsonObject.put("message","查询成功");
        }catch (SQLException e){
            log.error("查询数据失败"+ sql);
            jsonObject.put("success",false);
            jsonObject.put("message",e.getMessage());
        }finally {
            closeConn(conn,  null, null,statement);
            return jsonObject;
        }
    }

    /**
     * 查询数据
     * @param sql
     * @return
     */
    public List<LinkedHashMap<String,Object>> queryTableBySql(String sql){
        log.info("sql语句："+sql);
        List<LinkedHashMap<String,Object>> list = new ArrayList<>();
        Connection conn = null;
        Statement statement = null;
        try {
            conn = getConn();
            statement = conn.createStatement();
            ResultSet rs  = statement.executeQuery(sql);
            ResultSetMetaData rsmd = rs.getMetaData();
            while (rs.next()){
                LinkedHashMap<String,Object> map = new LinkedHashMap<>();
                for (int i = 1; i <= rsmd.getColumnCount(); i++) {
                    Object valueObj = rs.getObject(i);
                    valueObj = formatDataValue(valueObj);
                    map.put(rsmd.getColumnName(i),valueObj);
                }
                list.add(map);
            }
        }catch (SQLException e){
            log.error("查询数据失败"+ sql);
            throw new ServiceException("查询数据失败");
        }finally {
            closeConn(conn,  null, null,statement);
        }
        removeColumn(list);
        return list;
    }

    /**
     * 查询数据
     * @param sql
     * @return
     */
    public List<LinkedHashMap<String,Object>> queryTableBySqlNotRemove(String sql){
        log.info("sql语句："+sql);
        List<LinkedHashMap<String,Object>> list = new ArrayList<>();
        Connection conn = null;
        Statement statement = null;
        try {
            conn = getConn();
            statement = conn.createStatement();
            ResultSet rs  = statement.executeQuery(sql);
            ResultSetMetaData rsmd = rs.getMetaData();
            while (rs.next()){
                LinkedHashMap<String,Object> map = new LinkedHashMap<>();
                for (int i = 1; i <= rsmd.getColumnCount(); i++) {
                    Object valueObj = rs.getObject(i);
                    valueObj = formatDataValue(valueObj);
                    map.put(rsmd.getColumnName(i),valueObj);
                }
                list.add(map);
            }
        }catch (SQLException e){
            log.error("查询数据失败"+ sql);
            throw new ServiceException("查询数据失败");
        }finally {
            closeConn(conn,  null, null,statement);
        }
        return list;
    }

    /**
     * 删除数仓里面的表
     * @param tableName 表名称
     */
    public void dropTable(String tableName){
        Connection conn = null;
        PreparedStatement pstmt = null;
        try {
            conn = getConn();
            pstmt = conn.prepareStatement("TRUNCATE TABLE "+tableName);
            pstmt = conn.prepareStatement("drop table "+tableName);
            pstmt.execute();
        } catch (SQLException e) {
            log.error("删除表失败，表名称为："+tableName);
            throw new ServiceException("删除表失败");
        }finally {
            closeConn(conn, pstmt,null,null);
        }
    }

    /**
     * 清空表数据
     * @param tableName
     */
    public void emptyTable(String tableName){
        Connection conn = null;
        PreparedStatement pstmt = null;
        try {
            conn = getConn();
            pstmt = conn.prepareStatement("TRUNCATE TABLE "+tableName);
            pstmt.execute();
        } catch (SQLException e) {
            log.error("清空表失败，表名称为："+tableName);
//            throw new ServiceException("清空表失败");
        }finally {
            closeConn(conn, pstmt,null,null);
        }
    }


    public void insertData(String sql){
        RLock lock = redissonClient.getLock(RedisConstant.PERMISSION_LOCK);
        lock.lock();
        Connection conn = null;
        PreparedStatement pstmt = null;
        try {
            conn = getConn();
            pstmt = conn.prepareStatement(sql);
            pstmt.execute();
        } catch (SQLException e) {
            log.error("数据插入失败", e.getMessage());
            throw new ServiceException("数据插入失败");
        }finally{
            lock.unlock();
            closeConn(conn, pstmt, null,null);
        }
        log.debug("数据插入成功");
    }

    /**
     * 管理数据库使用
     * @param conn
     * @param pstmt
     * @param rs
     * @throws SQLException
     */
    private void closeConn(Connection conn, PreparedStatement pstmt, ResultSet rs,Statement statement){
        try{
            if (pstmt != null) {
                pstmt.close();
            }
            if (conn != null) {
                conn.close();
            }
            if (rs != null) {
                rs.close();
            }
            if (statement != null) {
                statement.close();
            }
        }catch (SQLException e) {
            log.error("关闭连接失败", e.getMessage());
            throw new ServiceException("关闭连接失败");
        }
    }

    /**
     * 移除同于数据列
     * @param list
     */
    public void removeColumn(List<LinkedHashMap<String,Object>> list){
        list.stream().peek(json -> json.remove(SystemConstant.DATABASEREDUNDANCYCOLUMN)).collect(Collectors.toList());
    }

}
